#![cfg_attr(debug_assertions, allow(dead_code, unused_imports, unused_variables, unused_macros))]

use actix_web::{get, post, web, App, HttpServer, HttpResponse, cookie::Cookie, HttpRequest};
use std::any::Any;
use std::sync::{Arc, Mutex};
use std::collections::HashMap;
use uuid::Uuid;
use std::fmt;
use std::fmt::Display;
use std::time::{Instant, Duration};
use serde_json::{json, Value};
use serde::{Serialize, Deserialize};
const DESC_MESSAGE: &str = "四则运算";
const LONGEST_SESSION_TIME: u64 = 1000 * 60 * 60 * 24 * 2; // 2天

#[derive(Debug, Copy, Clone, Serialize, Deserialize)]
pub enum TaskStatus {
    Created,
    Running,
    Completed,
    Failed,
    Stopped,
    Closed,
    ResultReceived,
    FailedToGetResult
}

/*
this service only allows one connection at a time
*/
struct AppState {
    sessions: Mutex<HashMap<String, Session>>,      // 保存所有挂载的客户端
    current_session: Mutex<Option<String>>,         // 未来支持同时进行多个客户端的工作，则需要修改
    workflow_num: Mutex<u32>,                       // 当前启动的工作流数量
    task_progress: Mutex<HashMap<String, (f32, TaskStatus)>>, // 任务进度
    task_results: Mutex<HashMap<String, Value>>,  // 任务结果
}

/*
客户端的一次挂载 */
#[derive(Clone)]
struct Session {
    user_name: String,      // 保存客户端的用户名
    status: UserStatus,     // 客户端的状态
    mount_time: Instant,    // 客户端连接到服务器的时间
}

/*
客户端一次挂载所处状态 */
#[derive(Clone)]
enum UserStatus {
    Mounted,        // 客户端连接到服务器
    Started,        // 客户端开始占用服务器进行工作
}

impl Display for UserStatus {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            UserStatus::Mounted => write!(f, "mounted"),
            UserStatus::Started => write!(f, "started"),
        }
    }
}

impl AppState {
    fn available_or_connected(&self, session_key: &String) -> bool {
        let ret = match *self.current_session.lock().unwrap() {
            Some(ref current_key) => current_key == session_key,
            _ => true
        };
        ret
    }

    fn get_session(&self, session_key: &String) -> Option<Session> {
        self.sessions.lock().unwrap().get(session_key).cloned()
    }

    fn set_session_status(&self, session_key: &String, status: UserStatus) -> Result<(), HttpResponse> {
        let mut sessions = self.sessions.lock().unwrap();
        if let Some(session) = sessions.get_mut(session_key) {
            session.status = status;
            Ok(())
        } else {
            Err(returns::unmounted())
        }
    }

    fn start_session(&self, session_key: &String) -> Result<(), HttpResponse> {
        let mut session = self.current_session.lock().unwrap();
        match *session {
            Some(_) => Err(returns::busy()),
            None => {
                session.replace(session_key.clone());
                Ok(())
            }
        }
    }

    fn stop_session(&self, session_key: &String) -> Result<(), HttpResponse> {
        let mut session = self.current_session.lock().unwrap();
        match *session {
            Some(ref current_key) => {
                if current_key == session_key {
                    if *self.workflow_num.lock().unwrap() > 0 {
                        Err(returns::working())
                    } else {
                        session.take();
                        Ok(())
                    }
                } else {
                    Err(returns::not_started())
                }
            }
            None => Err(returns::not_started()),
        }
    }
}

fn is_busy(state: &web::Data<Arc<AppState>>, key: &String) -> Result<(), HttpResponse> {
    let ret = if state.available_or_connected(key) {
        Ok(())
    } else {
        Err(returns::busy())
    };
    ret
}

fn cookie_check(body: &web::Json<Value>) -> Result<String, HttpResponse> {
    match body.get("session_key") {
        Some(v) => Ok(v.as_str().ok_or(returns::no_cookie())?.to_string()),
        None => Err(returns::no_cookie()),
    }
}

/*
开始一个工作 */
fn work_in(state: &web::Data<Arc<AppState>>, task_name: &str) {
    let mut workflow_num = state.workflow_num.lock().unwrap();
    *workflow_num += 1;
    log::info!("{}: {}", task_name, workflow_num);
}

/*
一个工作结束 */
fn work_out(state: &web::Data<Arc<AppState>>) {
    let mut workflow_num = state.workflow_num.lock().unwrap();
    *workflow_num -= 1;
}

/*
处理错误信息和ok的返回 */
#[macro_use]
mod returns {
    use actix_web::HttpResponse;
    use serde_json::json;
    const BUSY_MESSAGE: &str = "The server is busy. Only one user can be connected at a time.";
    
    macro_rules! build_json_response {
        ($resp:expr) => {
            $resp.0.body($resp.1.to_string())
        }
    }
                    
    macro_rules! ret_err {
        ($res:expr) => {
            match $res {
                Ok(v) => v,
                Err(e) => return e,
            }
        }
    }

    macro_rules! ret_actix_err {
        ($res:expr) => {
            match $res {
                Ok(v) => v,
                Err(e) => return HttpResponse::InternalServerError().body(format!("{:?}", e)),
            }
        }
    }
    
    pub fn no_cookie() -> HttpResponse {
        let mut resp = (HttpResponse::BadRequest(), json!({
            "error": "please provide a valid session key in the cookie",
        }));
        build_json_response!(resp)
    }
    
    pub fn unmounted() -> HttpResponse {
        let mut resp = (HttpResponse::BadRequest(), json!({
            "error": "user not mounted",
        }));
        build_json_response!(resp)
    }
    
    /*
    提供的session_key无效 */
    pub fn invalid_key() -> HttpResponse {
        let mut resp = (HttpResponse::BadRequest(), json!({
            "error": "please provide a valid session key",
        }));
        build_json_response!(resp)
    }
    
    pub fn running() -> HttpResponse {
        let mut resp = (HttpResponse::BadRequest(), json!({
            "error": "the session is running, please wait for it to finish or stop it first",
        }));
        build_json_response!(resp)
    }
    
    pub fn working() -> HttpResponse {
        let mut resp = (HttpResponse::BadRequest(), json!({
            "error": "a task is working in this session, please wait for it to finish or stop it first",
        }));
        build_json_response!(resp)
    }
    
    pub fn not_started() -> HttpResponse {
        let mut resp = (HttpResponse::BadRequest(), json!({
            "error": "the session has not been started",
        }));
        build_json_response!(resp)
    }
    
    pub fn busy() -> HttpResponse {
        let mut resp = (HttpResponse::TooManyRequests(), json!({
            "error": BUSY_MESSAGE,
        }));
        build_json_response!(resp)
    }
    
    pub fn already_mounted(key: &String) -> HttpResponse {
        let mut resp = (HttpResponse::BadRequest(), json!({
            "error": format!("service has already mounted, for session key: {}", &key),
        }));
        build_json_response!(resp)
    }

    pub fn ok() -> HttpResponse {
        let mut resp = (HttpResponse::Ok(), json!({
            "info": "ok",
        }));
        build_json_response!(resp)
    }

    pub fn invalid_input() -> HttpResponse {
        let mut resp = (HttpResponse::BadRequest(), json!({
            "error": "invalid input",
        }));
        build_json_response!(resp)
    }
}

/*
返回服务描述 */
#[get("/")]
async fn hello() -> String {
    DESC_MESSAGE.to_string()
}

#[get("/meta")]
async fn meta(req: HttpRequest, state: web::Data<Arc<AppState>>) -> HttpResponse {
    let mut resp = (HttpResponse::Ok(), json!({
        "name": "rust-calculator",
        "description": "this is a calculator service",
        "category": "test",
    }));
    build_json_response!(resp)
}

#[get("/tasks")]
async fn get_tasks(req: HttpRequest, state: web::Data<Arc<AppState>>) -> HttpResponse {
    let mut resp = (HttpResponse::Ok(), json!([
            "list_add_1: List<Int> -> List<Int>",
            "int_mul_2: Int -> Int",
        ]));
    build_json_response!(resp)
}

/*
清理超时挂载 */
fn clean_timeout_sessions(sessions: &mut HashMap<String, Session>) {
    let now = Instant::now();
    sessions.retain(|_, session| {
        now.duration_since(session.mount_time) < Duration::from_secs(LONGEST_SESSION_TIME)
    });
}

/*
挂载一个客户端，返回session_key */
#[get("/mount")]
async fn mount(req: HttpRequest, body: String, state: web::Data<Arc<AppState>>) -> HttpResponse {
    let user_name = body;
    let session_key = Uuid::new_v4().to_string();
    let mut sessions = state.sessions.lock().unwrap();
    sessions.insert(session_key.clone(), Session {
        user_name: user_name.clone(),
        status: UserStatus::Mounted,
        mount_time: Instant::now()
    });
    println!("user_name: {:?} mounted", user_name);
    clean_timeout_sessions(&mut sessions);
    let mut resp = (HttpResponse::Ok(), json!({
        "session_key": session_key,
        "healthy_info": "Healthy",
    }));
    build_json_response!(resp)
}

/*
卸载一个客户端 */
#[post("/unmount")]
async fn unmount(req: HttpRequest, body: web::Json<Value>, state: web::Data<Arc<AppState>>) -> HttpResponse {
    let key = ret_err!(cookie_check(&body));
    match state.get_session(&key) {
        None => returns::unmounted(),
        Some(ss) => match ss.status {
            UserStatus::Started => returns::running(),
            UserStatus::Mounted => {
                state.sessions.lock().unwrap().remove(&key);
                returns::ok()
            }
        }
    }
}

#[post("/task/list_add_1")]
async fn start_list_add_1(req: HttpRequest, state: web::Data<Arc<AppState>>, body: web::Json<Value>) -> HttpResponse {
    let key = ret_err!(cookie_check(&body));
    ret_err!(is_busy(&state, &key));

    let input_data = body;
    let task_id = Arc::new(Uuid::new_v4().to_string());

    let deal_func = |input_data: Value| {
        let input_data = input_data.as_array().unwrap();
        let mut result = vec![];
        for item in input_data {
            result.push(item.as_i64().unwrap_or(0) + 1);
        }
        json!(result)
    };

    // 启动异步任务
    let state_clone = state.clone();
    let task_id_clone = task_id.clone();
    std::thread::spawn(move || {
        process_task(task_id_clone, deal_func, input_data, state_clone);
    });

    HttpResponse::Ok().json(json!({ 
        "task_id": task_id.as_ref().clone(),
        "progress_pipe_addr": format!("/task/{}/progress", task_id.as_ref().clone()),
        "result_addr": format!("/task/{}/result", task_id.as_ref().clone()),
     }))
}

#[post("/task/int_mul_2")]
async fn start_int_mul_2(req: HttpRequest, state: web::Data<Arc<AppState>>, body: web::Json<Value>) -> HttpResponse {
    let key = ret_err!(cookie_check(&body));
    ret_err!(is_busy(&state, &key));

    let input_data = body;
    let task_id = Arc::new(Uuid::new_v4().to_string());

    let deal_func = |input_data: Value| {
        let input_data = input_data.as_i64().unwrap();
        json!(input_data * 2)
    };

    // 启动异步任务
    let state_clone = state.clone();
    let task_id_clone = task_id.clone();
    std::thread::spawn(move || {
        process_task(task_id_clone, deal_func, input_data, state_clone);
    });

    HttpResponse::Ok().json(json!({ 
        "task_id": task_id.as_ref().clone(),
        "progress_pipe_addr": format!("/task/{}/progress", task_id.as_ref().clone()),
        "result_addr": format!("/task/{}/result", task_id.as_ref().clone()),
     }))
}

type StateType = web::Data<Arc<AppState>>;

// 处理任务的函数
fn process_task(task_id: Arc<String>, deal_func: fn(Value) -> Value, input_data: Value, state: StateType) {
    {
        let mut progress = state.task_progress.lock().unwrap();
        progress.insert(task_id.as_ref().clone(), (0.0, TaskStatus::Created));
    }
    println!("task start: {}", task_id);
    println!("input_data: {:?}", input_data);

    // 模拟任务处理
    for i in 1..=5 {
        std::thread::sleep(Duration::from_secs(1));
        let mut progress = state.task_progress.lock().unwrap();
        progress.insert(task_id.as_ref().clone(), (i as f32 * 20.0, TaskStatus::Running));
        println!("progress: {:?}", progress);
    }

    let result = deal_func(input_data);
    println!("result: {:?}", result); 
    {
        let mut progress = state.task_progress.lock().unwrap();
        progress.insert(task_id.as_ref().clone(), (100.0, TaskStatus::Completed));
        let mut results = state.task_results.lock().unwrap();
        results.insert(task_id.as_ref().clone(), result);
    }
}

// 获取任务进度
#[get("/task/{task_id}/progress")]
async fn get_progress(task_id: web::Path<String>, state: web::Data<Arc<AppState>>) -> HttpResponse {
    let progress = state.task_progress.lock().unwrap();
    if let Some(&status) = progress.get(&task_id.into_inner()) {
        HttpResponse::Ok().json(json!({ 
            "percentage": status.0,
            "status": status.1
        }))
    } else {
        HttpResponse::NotFound().json(json!({ "error": "Task not found" }))
    }
}

// 获取任务结果
#[get("/task/{task_id}/result")]
async fn get_result(task_id: web::Path<String>, state: web::Data<Arc<AppState>>) -> HttpResponse {
    let results = state.task_results.lock().unwrap();
    if let Some(result) = results.get(&task_id.into_inner()) {
        println!("return task result: {:?}", result);
        HttpResponse::Ok().json(result)
    } else {
        HttpResponse::NotFound().json(json!({ "error": "Task not found" }))
    }
}

/*
启动服务器 */
#[actix_web::main]
async fn main() -> std::io::Result<()> {
    // create a shared state
    let state = Arc::new(AppState {
        sessions: Mutex::new(HashMap::new()),
        current_session: Mutex::new(None),
        workflow_num: Mutex::new(0),
        task_progress: Mutex::new(HashMap::new()),
        task_results: Mutex::new(HashMap::new()),
    });

    env_logger::init();
    log::info!("starting server at 127.0.0.1:8102");

    HttpServer::new(move || {
        App::new()
            .app_data(web::Data::new(state.clone()))
            .service(hello)
            .service(meta)
            .service(get_tasks)
            .service(mount)
            .service(unmount)
            .service(start_list_add_1)
            .service(start_int_mul_2)
            .service(get_progress)
            .service(get_result)
    })
    .bind("127.0.0.1:8102")?
    .run()
    .await
}
